package cn.itcast.kafka.optimize;


import cn.itcast.kafka.config.KafkaProducerConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 优化后的Kafka生产者程序（使用外部配置文件）
 */
public class KafkaProducerTestOptimize {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 通过配置工具类获取Kafka配置（不再硬编码）
        Properties props = KafkaProducerConfig.getProducerProperties();

        // 2. 创建生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);

        int MAX = 10000000;

        // 3. 发送消息
        for(int i = 1000000; i < MAX; ++i) {
            // 异步回调方式发送消息
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test1", null, i + "");
            kafkaProducer.send(producerRecord, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception == null) {
                        System.out.printf("topic:%s 分区id：%d 偏移量：%d%n",
                                metadata.topic(), metadata.partition(), metadata.offset());
                    } else {
                        System.err.println("生产消息出现异常：" + exception.getMessage());
                        exception.printStackTrace();
                    }
                }
            });

            Thread.sleep(1000);
        }

        // 4. 关闭生产者
        kafkaProducer.close();
    }
}
